package com.ihr360.binlog.component;


import com.alibaba.otter.canal.protocol.CanalEntry;
import com.ihr360.binlog.dao.SyncedBasicStaffInfoDao;
import com.ihr360.binlog.model.EventRow;
import com.ihr360.commons.lang.CollectionUtils;
import com.ihr360.commons.lang.DateUtils;
import com.ihr360.commons.lang.StringUtils;
import com.ihr360.shared.basicstaffinfo.dto.SyncedBasicStaffInfo;
import com.ihr360.type.StaffStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import rx.functions.Action1;

import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Vector;
import java.util.stream.Collectors;

/**
 * 订阅tab_staff_info 数据变化
 * Created by davidwei on 17-7-21.
 */
@Component
public class StaffInfoEventEmitter {
    Logger logger = LoggerFactory.getLogger(StaffInfoEventEmitter.class.getName());

    @Value("${ihr360.tab_staff_info.sync.by.lastupdate:false}")
    private boolean syncIfLastUpdateChanged;

    @Autowired
    SyncedBasicStaffInfoDao syncedBasicStaffInfoDao;

    final List<SyncedBasicStaffInfo> infos = new Vector<>();

    private int count = 0;
    private long lastUpdate = 0;
    public Action1<EventRow> onEventRow = new Action1<EventRow>() {
        @Override
        public void call(EventRow syncRow) {
            CanalEntry.RowChange rowChange = syncRow.getRowChange();
            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                CanalEntry.EventType eventType = rowChange.getEventType();
                if (eventType == CanalEntry.EventType.DELETE) {
                    Optional<CanalEntry.Column> idOptional = rowData.getBeforeColumnsList().stream().filter(c -> c.getName().equals("id")).findFirst();
                    if (idOptional.isPresent()) {
                        CanalEntry.Column column = idOptional.get();
                        String id = column.getValue();
                        syncedBasicStaffInfoDao.delete(id);
                    }
                } else if (eventType == CanalEntry.EventType.INSERT) {
                    List<CanalEntry.Column> canalColumns = rowData.getAfterColumnsList();
                    updateChanges2Dest(canalColumns);
                    continue;
                } else if (eventType == CanalEntry.EventType.UPDATE) {
                    List<CanalEntry.Column> canalColumns = rowData.getAfterColumnsList();
                    updateChanges2Dest(canalColumns);
                }
            }
        }
    };

    private void updateChanges2Dest(List<CanalEntry.Column> canalColumns) {
        if (CollectionUtils.isEmpty(canalColumns)) {
            return;
        }

        Map<String, CanalEntry.Column> columnMap = canalColumns.stream().collect(Collectors.toMap(CanalEntry.Column::getName, c -> c));

        CanalEntry.Column companyIdCol = columnMap.get(SyncedBasicStaffInfo.Field.companyId);
        CanalEntry.Column departmentIdCol = columnMap.get(SyncedBasicStaffInfo.Field.departmentId);
        CanalEntry.Column departmentNameCol = columnMap.get(SyncedBasicStaffInfo.Field.departmentName);
        CanalEntry.Column enrollInDateCol = columnMap.get(SyncedBasicStaffInfo.Field.enrollInDate);
        CanalEntry.Column idCol = columnMap.get(SyncedBasicStaffInfo.Field.id);
        CanalEntry.Column staffNameCol = columnMap.get(SyncedBasicStaffInfo.Field.staffName);
        CanalEntry.Column idCardCol = columnMap.get(SyncedBasicStaffInfo.Field.idCardNo);
        CanalEntry.Column idCardType = columnMap.get(SyncedBasicStaffInfo.Field.idCardType);
        CanalEntry.Column mobileNoCol = columnMap.get(SyncedBasicStaffInfo.Field.mobileNo);
        CanalEntry.Column sexCol = columnMap.get(SyncedBasicStaffInfo.Field.sex);
        CanalEntry.Column staffStatusCol = columnMap.get(SyncedBasicStaffInfo.Field.staffStatus);
        CanalEntry.Column lastWorkDayCol = columnMap.get(SyncedBasicStaffInfo.Field.lastWorkDay);

        CanalEntry.Column positionNameCol = columnMap.get(SyncedBasicStaffInfo.Field.positionName);
        CanalEntry.Column positionIdCol = columnMap.get(SyncedBasicStaffInfo.Field.positionId);
        CanalEntry.Column staffNoCol = columnMap.get(SyncedBasicStaffInfo.Field.staffNo);

        boolean isUpdate =
                isUpdate(staffStatusCol) ||
                        isUpdate(departmentIdCol) ||
                        isUpdate(departmentNameCol) ||
                        isUpdate(enrollInDateCol) ||
                        isUpdate(idCardCol) ||
                        isUpdate(idCardType) ||
                        isUpdate(mobileNoCol) ||
                        isUpdate(sexCol) ||
                        isUpdate(companyIdCol) ||
                        isUpdate(idCol) ||
                        isUpdate(lastWorkDayCol) ||
                        isUpdate(positionNameCol) ||
                        isUpdate(positionIdCol) ||
                        isUpdate(staffNoCol);

        if (!isUpdate && !syncIfLastUpdateChanged) {
            return;
        }
        SyncedBasicStaffInfo info = new SyncedBasicStaffInfo();
        info.setCompanyId(getValue(companyIdCol));
        info.setDepartmentId(getLongValue(departmentIdCol));
        info.setDepartmentName(getValue(departmentNameCol));
        info.setSex(getValue(sexCol));
        info.setMobileNo(getValue(mobileNoCol));
        info.setIdCardType(getValue(idCardType));
        info.setStaffStatus(getStaffStatus(staffStatusCol));
        info.setIdCardNo(getValue(idCardCol));
        info.setStaffName(getValue(staffNameCol));
        info.setId(getValue(idCol));
        info.setEnrollInDate(getDate(enrollInDateCol));
        info.setLastWorkDay(getDate(lastWorkDayCol));
        info.setPositionId(getValue(positionIdCol));
        info.setPositionName(getValue(positionNameCol));
        info.setStaffNo(getValue(staffNoCol));
        infos.add(info);
        count++;

        long now = Calendar.getInstance().getTime().getTime();
        if (count >= 1000 || (now - lastUpdate > 1000)) {
            long start = lastUpdate = Calendar.getInstance().getTime().getTime();
            syncedBasicStaffInfoDao.update(infos);
            long end = Calendar.getInstance().getTime().getTime();
            logger.info("updated cnt=" + count);
            logger.info("---update time cost=" + (end - start) / 1000);
            infos.clear();
            count = 0;
            lastUpdate = Calendar.getInstance().getTime().getTime();
        }

    }

    private String getValue(CanalEntry.Column column) {
        return column != null ? column.getValue() : null;
    }

    private Date getDate(CanalEntry.Column column) {
        return column != null && StringUtils.isNotEmpty(column.getValue()) ? DateUtils.dateFormat(column.getValue(), DateUtils.STANDARD_DATE_FORMAT_YMD) : null;
    }

    private boolean isUpdate(CanalEntry.Column column) {
        return column != null ? column.getUpdated() : false;
    }

    private Long getLongValue(CanalEntry.Column column) {
        return column != null ? (StringUtils.isNotEmpty(column.getValue()) ? Long.valueOf(column.getValue()) : null) : null;
    }

    private StaffStatus getStaffStatus(CanalEntry.Column column) {
        return column != null ? StaffStatus.valueOf(column.getValue()) : null;
    }


}
